package org.xx.armory.rpc;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 用于执行服务的执行器。
 */
public class ServiceExecutor
        extends ThreadPoolExecutor {
    /**
     * 核心线程数占最大并发数的比例。
     */
    private static final float CORE_POOL_FACTOR = .8F;
    private final Object workingThreadLock;
    private final List<Thread> workingThreads;

    /**
     * 构造执行器实例。
     *
     * @param parallelism
     *         最大并发数。
     */
    public ServiceExecutor(
            int parallelism
    ) {
        super((int) (parallelism * CORE_POOL_FACTOR), parallelism, 0L, TimeUnit.MICROSECONDS, new LinkedBlockingQueue<>());

        this.workingThreadLock = new Object();
        this.workingThreads = new ArrayList<>(32);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void beforeExecute(
            Thread t,
            Runnable r
    ) {
        super.beforeExecute(t, r);

        synchronized (this.workingThreadLock) {
            final ListIterator<Thread> iterator = this.workingThreads.listIterator();
            boolean found = false;
            while (iterator.hasNext()) {
                final Thread c = iterator.next();
                if (!c.isAlive()) {
                    iterator.remove();
                } else if (c.equals(t)) {
                    found = true;
                    break;
                }
            }
            if (!found) {
                iterator.add(t);
            }
        }
    }

    /**
     * 获取当前的线程集合。
     *
     * @return 当前的线程集合。
     */
    public List<Thread> getWorkThreads() {
        synchronized (this.workingThreadLock) {
            return Collections.unmodifiableList(this.workingThreads);
        }
    }
}
